package org.dna.mqtt.moquette.processor.impl;

import java.nio.ByteBuffer;

import org.dna.mqtt.moquette.messaging.spi.impl.ValueEvent;
import org.dna.mqtt.moquette.processor.impl.MqttProtocolProcessor.WillMessage;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.ServerChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Handles CONNECT type protocol
 * @author andrea, puran
 */
public class ConnectProcessor extends AbstractProtocolProcessor {
	private static final Logger LOG = LoggerFactory.getLogger(ConnectProcessor.class);
	private static final byte PROTOCOL_VERSION_3 = 0x03;
    
	public ConnectProcessor(ServerChannel session, ConnectMessage message) {
		super(session, message);
	}

	@Override
	public void processEvent(ValueEvent valueEvent) {
        processConnect();
	}
	
	private void processConnect() {
		ServerChannel session = getSession(); 
		ConnectMessage msg = (ConnectMessage)getMessage();
		
        LOG.debug("processConnect for client {}", msg.getClientID());
        
        MqttProtocolProcessor baseProcessor = getBaseProcessor();
        if (msg.getProcotolVersion() != PROTOCOL_VERSION_3) {
            ConnAckMessage badProto = new ConnAckMessage();
            badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
            LOG.warn("processConnect sent bad proto ConnAck");
            session.write(badProto);
            session.close(false);
            return;
        }

        if (msg.getClientID() == null || msg.getClientID().length() > 23) {
        	LOG.debug("Client id too long "+msg.getClientID());
            ConnAckMessage okResp = new ConnAckMessage();
            okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
            session.write(okResp);
            return;
        }

        //if an old client with the same ID already exists close its session.
        if (baseProcessor.clientExists(msg.getClientID())) {
            //clean the subscriptions if the old used a cleanSession = true
            ServerChannel oldSession = getBaseProcessor().getClientSession(msg.getClientID());
            boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION);
            if (cleanSession) {
                //cleanup topic subscriptions
            	getBaseProcessor().cleanSession(msg.getClientID());
            }
            oldSession.close(false);
        }

        ConnectionDescriptor connectionDescriptor = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession());
        baseProcessor.addClient(msg.getClientID(), connectionDescriptor);

        int keepAlive = msg.getKeepAlive();
        LOG.debug("Connect with keepAlive {} s",  keepAlive);
        session.setAttribute(Constants.KEEP_ALIVE, keepAlive);
        session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession());
        //used to track the client in the subscription and publishing phases.
        session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID());

        session.setIdleTime(Math.round(keepAlive * 1.5f));

        //Handle will flag
        if (msg.isWillFlag()) {
            AbstractMessage.QOSType willQos = AbstractMessage.QOSType.values()[msg.getWillQos()];
            byte[] willPayload = msg.getWillMessage().getBytes();
            ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip();
            
            //save the will testment in the clientID store
            WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos );
            baseProcessor.addWillMessage(msg.getClientID(), will);
        }

        //handle user authentication
        if (msg.isUserFlag()) {
            String pwd = null;
            if (msg.isPasswordFlag()) {
                pwd = msg.getPassword();
            }
            if (!baseProcessor.checkIfClientValid(msg.getUsername(), pwd)) {
                ConnAckMessage okResp = new ConnAckMessage();
                okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
                session.write(okResp);
                return;
            }
        }
        
        baseProcessor.activeClient(msg.getClientID());
        //handle clean session flag
        if (msg.isCleanSession()) {
            //remove all prev subscriptions
            //cleanup topic subscriptions
        	baseProcessor.cleanSession(msg.getClientID());
        }

        ConnAckMessage okResp = new ConnAckMessage();
        okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
        LOG.debug("processConnect sent OK ConnAck");
        session.write(okResp);
        LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession());
        
        if (!msg.isCleanSession()) {
            //force the republish of stored QoS1 and QoS2
        	baseProcessor.republishStored(msg.getClientID());
        }
    }
}
